package com.dhy.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;

import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;

/**
 * @author 大忽悠
 * @create 2022/12/27 9:36
 */
@Slf4j
@RequiredArgsConstructor
public class CanalManager implements InitializingBean, ApplicationRunner {
    private final CanalProperties canalProperties;
    private final CanalMessageDispatcher canalMessageDispatcher;
    private CanalConnector canalConnector;

    @Override
    public void afterPropertiesSet() throws Exception {
        canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalProperties.getHost(), canalProperties.getPort()), canalProperties.getDestination(), "", "");
        log.info("单体版canalConnector初始化完毕,主机: {} ,端口: {} ,地址: {}", canalProperties.getHost(), canalProperties.getPort(), canalProperties.getDestination());
        canalMessageDispatcher.setUnderlineToCamel(canalProperties.getUnderlineToCamel());
        log.info("是否开启数据库列名到JavaBean属性名的自动驼峰转换: {}",canalProperties.getUnderlineToCamel());
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        //打开连接
        canalConnector.connect();
        //订阅数据库表,全部表
        canalConnector.subscribe(".*\\..*");
        //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
        canalConnector.rollback();
        //开始拉取数据
        CompletableFuture.runAsync(this::doFetch, Executors.newSingleThreadExecutor());
    }

    private void doFetch() {
        while (true) {
            try {
                // 获取指定数量的数据
                Message message = canalConnector.getWithoutAck(canalProperties.getFetchSize());
                //获取批量ID
                long batchId = message.getId();
                //获取批量的数量
                int size = message.getEntries().size();
                //如果没有数据
                if (batchId == -1 || size == 0) {
                    //线程休眠
                    Thread.sleep(Duration.ofSeconds(canalProperties.getFetchInterval()).toMillis());
                } else {
                    //如果有数据,处理数据
                    canalMessageDispatcher.dispatchMessage(message);
                }
                //进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
                canalConnector.ack(batchId);
            } catch (Throwable t) {
                t.printStackTrace();
            }
        }
    }


}
